package streamforce.crypto.cloud.subscriber;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import streamforce.crypto.KPAbe;
import streamforce.crypto.cloud.EPCloud;
import streamforce.crypto.cloud.EPCloudServer;
import streamforce.crypto.cloud.TimeTuple;
import streamforce.crypto.cloud.Tuple;
import streamforce.crypto.event.Ciphertext;
import streamforce.crypto.event.TransformedCTEventAll;
import streamforce.crypto.policies.Policy;

//for query: select sum(price) from StockEvent(cond):win_batch_length(ws)
//assuming that the cryptohandler has been initialized
public class CloudSumAllAgg1 extends CloudSumPriceAgg1{
	final Log log = LogFactory.getLog(CloudSumAllAgg1.class);			
	long totalTime = 0; 
	byte[][] tmpOriginalPrice = null, tmpOriginalVolume = null,
			tmpTransformedPrice = null, tmpTransformedVolume = null;
	
	public CloudSumAllAgg1(int pi, int windowSize, KPAbe ch){
		super(pi,windowSize, ch);		
	}	
	
	private void processPrice(byte[] ciphertext){
		byte[] original= this.cryptoHandler.extractOriginalCipher(this.streamId, ciphertext);
		byte[] transformed=this.cryptoHandler.transform_simple(ciphertext, this.streamId, this.userId);
		if (this.acc%windowSize==0){			
			this.tmpOriginalPrice = new byte[windowSize][original.length];			
			this.tmpTransformedPrice = new byte[windowSize][transformed.length];			
		}		
		this.tmpOriginalPrice[acc%windowSize] = original;
		this.tmpTransformedPrice[acc%windowSize] = transformed;							
	}
	
	private void processVolume(byte[] ciphertext){
		byte[] original= this.cryptoHandler.extractOriginalCipher(this.streamId, ciphertext);
		byte[] transformed=this.cryptoHandler.transform_simple(ciphertext, this.streamId, this.userId);
		if (this.acc%windowSize==0){			
			this.tmpOriginalVolume = new byte[windowSize][original.length];			
			this.tmpTransformedVolume = new byte[windowSize][transformed.length];			
		}		
		this.tmpOriginalVolume[acc%windowSize] = original;
		this.tmpTransformedVolume[acc%windowSize] = transformed;							
	}
	
	public  void update(int timestamp, int hour, int stockId, Ciphertext ctPrice, Ciphertext ctVolume) {
		TimeTuple tt = EPCloudServer.tupleTime.get(new Tuple(stockId, timestamp)); 
		if (tt.getCount()==0)
			this.statTuple.update(); 			
		tt.increaseCount();
		
		byte[] ciphertextPrice = ctPrice.getCipher();
		byte[] ciphertextVolume = ctVolume.getCipher();
		
		long start = System.currentTimeMillis(); 
		processPrice(ciphertextPrice);
		processVolume(ciphertextVolume);		
		long end = System.currentTimeMillis(); 
		this.totalTime += (end-start); 
		this.stat.update();
		
		this.acc++;
		// then construct and send another event, only when the window is filled
		if ((this.acc >0) && (this.acc % windowSize == 0)) {	
			start = System.currentTimeMillis(); 
			byte[] pop = this.cryptoHandler.multiplyCiphers(this.tmpOriginalPrice, this.streamId);
			byte[] pov = this.cryptoHandler.multiplyCiphers(this.tmpOriginalVolume, streamId);
			byte[] ptp = this.cryptoHandler.multiplyCiphers(this.tmpTransformedPrice, this.streamId);
			byte[] ptv = this.cryptoHandler.multiplyCiphers(this.tmpTransformedVolume, this.streamId);
			byte[][] o = new byte[2][pop.length];
			o[0] = pop; o[1] = pov;
			byte[][] t = new byte[2][ptp.length];
			t[0] = ptp; t[1] = ptv;
			TransformedCTEventAll cte = new TransformedCTEventAll(
					this.userId, timestamp, hour, stockId, o, t,
					TransformedCTEventAll.SELECT_ALL);
			EPCloud.getEPServiceProvider().getEPRuntime().sendEvent(cte);
			end = System.currentTimeMillis();
			totalTime+=(end-start); 
			log.info("agg1 transform time = "+totalTime);
			this.totalTime = 0; 
		}		
	}
	
}
